package com.van;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;

/**
 * @Author van.shu
 * @Date 2020/4/25 16:30
 */

public class HDFSClient {


    FileSystem fs = null;
    @Before
    public void before() throws IOException, InterruptedException {
        Configuration configuration = new Configuration();

        fs = FileSystem.get(URI.create("hdfs://hadoop101:9000"), new Configuration(), "bigdata");

    }


    @Test
    public void mkdir() throws IOException {

        fs.mkdirs(new Path("/scj/10/"));
        fs.close();

    }

    @Test
    public void copyFromLocal() throws IOException {
        fs.copyFromLocalFile(new Path("D:/hello.txt"), new Path("/scj/04/"));
        fs.close();
    }

    @Test
    public void copyToLocal() throws IOException {
        fs.copyToLocalFile(false, new Path("/hello.txt"), new Path("d:/b.txt"), true);
        fs.close();

    }

    @Test
    public void delete() throws IOException {

        //true: 递归删除文件和文件夹
        fs.delete(new Path("/scj"), true);
        fs.close();

    }


    @Test
    public void rename() throws IOException {
        fs.rename(new Path("/hello.txt"), new Path("/world.txt"));
        fs.close();
    }

    /**
     * 文件详情查看
     * @throws IOException
     */
    @Test
    public void listFiles() throws IOException {
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println(fileStatus.getPath().getName());
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getGroup());
            System.out.println(fileStatus.getPermission());

            BlockLocation[] blockLocations = fileStatus.getBlockLocations();

            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println("host = " + host);
                }
            }
        }

        fs.close();
    }

    /**
     * 判断文件和文件夹
     */
    @Test
    public void listStatus() throws IOException {

        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));

        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isFile()) {
                System.out.println("file:"+fileStatus.getPath().getName());
            }else {
                System.out.println("dictionary:" + fileStatus.getPath().getName());

            }
        }
    }

    /**
     * 文件上传到HDFS
     * @throws IOException
     */
    @Test
    public void putFileToHdfs() throws IOException {

        //输入流
        FileInputStream fis = new FileInputStream("d:/hello.txt");


        //输出流
        FSDataOutputStream fos = fs.create(new Path("/van.txt"));


        //流拷贝
        IOUtils.copyBytes(fis, fos, new Configuration());

        //关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);

    }

    /**
     * 从HDFS上下载文件
     * @throws IOException
     */
    @Test
    public void getFileFromHDFS() throws IOException {

        FSDataInputStream fis = fs.open(new Path("/van.txt"));

        FileOutputStream fos = new FileOutputStream("./xx.txt");

        IOUtils.copyBytes(fis, fos, new Configuration());

        //关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    }





}

